package cn.tedu.stream.checkpoint

import org.apache.flink.api.common.restartstrategy.RestartStrategies
import org.apache.flink.api.common.time.Time
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment

object StreamRestartDemo {

  def main(args: Array[String]): Unit = {

    //环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //开启检查点
    env.enableCheckpointing(5000)
    //设置检查点的持久化
    env.setStateBackend(new FsStateBackend("hdfs://hadoop01:8020/ck/05"))
    //取消任务保存检查点数据
    env.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

    //设置重启策略
    //(1)固定延迟重启策略
    //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.seconds(10)))

    //(2)失败率重启策略
    env.setRestartStrategy(RestartStrategies.failureRateRestart(3,Time.minutes(10),Time.seconds(10)))

    //(3)无重启策略:默认就是无重启策略
    //env.setRestartStrategy(RestartStrategies.noRestart())


    //数据源 hello world hello flink
    val source = env.socketTextStream("hadoop01", 9999)

    import org.apache.flink.api.scala._
    val result = source.flatMap(_.split(" ")).map(word => {
      if ("laowang".equals(word)) {
        throw new RuntimeException("老王来了，程序重启！！！！")
      }
      (word, 1)
    }).keyBy(0).sum(1)

    result.print()
    env.execute()
  }

}
